package com.kc.gb.device.service.impl;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson2.JSON;
import com.kc.gb.device.common.adapter.DeviceAdaptor;
import com.kc.gb.device.common.contants.MQConst;
import com.kc.gb.device.common.dto.I18NException;
import com.kc.gb.device.common.enums.ErrorCode;
import com.kc.gb.device.common.properties.DeviceProperties;
import com.kc.gb.device.events.support.EmergencyEvent;
import com.kc.gb.device.service.DeviceMQService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;

@Slf4j
@Service
public class DeviceMQServiceImpl implements DeviceMQService {

    @Resource
    DeviceProperties deviceProperties;

    @Resource
    DeviceAdaptor deviceAdaptor;

    private static DefaultMQProducer ptlCallbackProducer;
    private static DefaultMQPushConsumer ptlCommandConsumer;

    @Override
    public void sendFeedBackMessage(EmergencyEvent event) throws Exception {
        // 发送 MQ 消息
        String stationCode = deviceAdaptor.getApplicationStationCode();
        if(StringUtils.isEmpty(stationCode)){
            throw new I18NException(ErrorCode.STATION_NOT_BIND);
        }
        Message message = new Message();
        message.setTopic(MQConst.Device.device_to_station_topic);
        message.setKeys(StrUtil.format("Emergency-{}-{}",stationCode,event.isSuspendSystem()));
        message.setBody(JSON.toJSONString(event).getBytes(StandardCharsets.UTF_8));
        message.setTags(stationCode);
        SendResult sendResult = ptlCallbackProducer.send(message);
        log.info("Sending Emergency Result:{}", sendResult);
    }

    @Override
    public DefaultMQProducer createCommandProducer() throws Exception {
        String station = deviceAdaptor.getApplicationStationCode();
        String topic = MQConst.Device.device_to_station_topic;
        String producerGroup = MQConst.Device.device_to_station_topic_producer;
        String nameServer = getNameServer();
        // 创建生产者
        if(StringUtils.isNotEmpty(station)){
            log.info("MQProducer {} 正在创建.... ",topic);
            DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
            producer.setNamesrvAddr(nameServer);
            producer.setVipChannelEnabled(false);
            //producer.setMaxMessageSize(maxMessageSize);
            //producer.setSendMsgTimeout(sendMsgTimeOut);
            producer.setRetryTimesWhenSendAsyncFailed(10);
            producer.start();
            ptlCallbackProducer = producer;
            return producer;
        }
        return null;
    }

    private String getNameServer(){
        return StrUtil.format( "{}:9876",deviceProperties.getHostIP());
    }
}
